package org.rcy.framework.file.io;

import com.google.common.collect.Lists;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;

import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author 吴康桥
 * @Description
 * @date 2022/12/13 16:11
 */


public class IOUtils {

	public static void concurrentReadBigFileWithThreadPool(String pathName, ProcessCallBack callBack) throws IOException, ExecutionException, InterruptedException {
		// 创建一个 最大线程数为 10，队列最大数为 100 的线程池
		ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 10, 60l, TimeUnit.SECONDS, new LinkedBlockingDeque<>(100));
		// 使用 Apache 的方式逐行读取数据
		LineIterator fileContents = FileUtils.lineIterator(new File(pathName), StandardCharsets.UTF_8.name());
		List<String> lines = Lists.newArrayList();
		while (fileContents.hasNext()) {
			String nextLine = fileContents.nextLine();
			lines.add(nextLine);
			// 读取到十万的时候
			if (lines.size() == 100000) {
				// 拆分成两个 50000 ，交给异步线程处理
				List<List<String>> partition = Lists.partition(lines, 50000);
				List<Future> futureList = Lists.newArrayList();
				for (List<String> strings : partition) {
					Future<?> future = threadPoolExecutor.submit(() -> {
						if (callBack != null) {
							callBack.callBack(strings);
						}
					});
					futureList.add(future);
				}
				// 等待两个线程将任务执行结束之后，再次读取数据。这样的目的防止，任务过多，加载的数据过多，导致 OOM
				for (Future future : futureList) {
					// 等待执行结束
					future.get();
				}
				// 清除内容
				lines.clear();
			}


		}
		// lines 若还有剩余，继续执行结束
		if (!lines.isEmpty()) {
			// 继续执行
			if (callBack != null) {
				callBack.callBack(lines);
			}
		}
		threadPoolExecutor.shutdown();
	}

}
